package com.dhy.canal;

import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.dhy.canal.util.CustomLocalDateTimeConvert;
import com.dhy.canal.util.StringUtil;
import lombok.Builder;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ObjectUtils;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.BeanWrapperImpl;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.core.GenericTypeResolver;
import org.springframework.core.convert.support.GenericConversionService;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author 大忽悠
 * @create 2022/12/27 9:54
 */
@Slf4j
public class CanalMessageDispatcher implements BeanPostProcessor, BeanFactoryAware {
    private final Map<String, List<CanalMessageListener<Object>>> canalMessageListenerMap = new HashMap<>();
    private GenericConversionService conversionService;
    private Boolean underlineToCamel=Boolean.TRUE;

    public void dispatchMessage(Message message) {
        Map<String, CanalEventHolder> canalEventHolderMap = CanalMessageParser.parse(message);
        canalEventHolderMap.keySet().forEach(tableName -> {
                    //定位监听当前表的listener
                    List<CanalMessageListener<Object>> canalMessageListeners = canalMessageListenerMap.get(tableName);
                    if (!ObjectUtils.isEmpty(canalMessageListeners)) {
                        doDispatch(canalMessageListeners, canalEventHolderMap.get(tableName));
                    }
                }
        );
    }

    private void doDispatch(List<CanalMessageListener<Object>> canalMessageListeners, CanalEventHolder canalEventHolder) {
        for (CanalMessageListener<Object> canalMessageListener : canalMessageListeners) {
            //拿到监听器上的泛型类型
            GenericTypeResolver.getTypeVariableMap(canalMessageListener.getClass()).forEach(
                    (k, v) -> {
                        //实例化泛型
                        try {
                            //如果是删除语句
                            if (canalEventHolder.eventType == CanalEntry.EventType.DELETE) {
                                doHandleDeleteEvent(canalEventHolder, canalMessageListener, (Class<?>) v);
                            } else if (canalEventHolder.eventType == CanalEntry.EventType.INSERT) {
                                doHandleInsertEvent(canalEventHolder, canalMessageListener, (Class<?>) v);
                            } else {
                                doHandleUpdateEvent(canalEventHolder, canalMessageListener, (Class<?>) v);
                            }
                        } catch (Exception e) {
                            throw new RuntimeException("canal事件派发过程中出现异常",e);
                        }
                    }
            );

        }
    }

    private void doHandleUpdateEvent(CanalEventHolder canalEventHolder, CanalMessageListener<Object> canalMessageListener, Class<?> v) throws InstantiationException, IllegalAccessException {
        canalMessageListener.update(getAutowiredBeanInstance(v, canalEventHolder.afterColumns), getAutowiredBeanInstance(v, canalEventHolder.beforeColumns));
    }

    private void doHandleInsertEvent(CanalEventHolder canalEventHolder, CanalMessageListener<Object> canalMessageListener, Class<?> v) throws InstantiationException, IllegalAccessException {
        canalMessageListener.insert(getAutowiredBeanInstance(v, canalEventHolder.afterColumns));
    }

    private void doHandleDeleteEvent(CanalEventHolder canalEventHolder, CanalMessageListener<Object> canalMessageListener, Class<?> v) throws InstantiationException, IllegalAccessException {
        canalMessageListener.delete(getAutowiredBeanInstance(v, canalEventHolder.beforeColumns));
    }

    private Object getAutowiredBeanInstance(Class<?> v, List<CanalEntry.Column> canalEventHolder) throws InstantiationException, IllegalAccessException {
        BeanWrapperImpl record = new BeanWrapperImpl(v.newInstance());
        record.setConversionService(conversionService);
        for (CanalEntry.Column column : canalEventHolder) {
            record.setPropertyValue(underlineToCamel?StringUtil.underlineToCamel(column.getName()):column.getName(), column.getValue());
        }
        return record.getWrappedInstance();
    }

    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
        if (!isCanalMessageListener(bean)) {
            return bean;
        }
        String listenTable = CanalMessageParser.parseListenTable(bean);
        canalMessageListenerMap.computeIfAbsent(listenTable, table -> {
            List<CanalMessageListener<Object>> canalMessageListeners = new ArrayList<>();
            canalMessageListeners.add((CanalMessageListener<Object>) bean);
            return canalMessageListeners;
        });
        return bean;
    }

    private boolean isCanalMessageListener(Object bean) {
        //如果bean被代理了,需要拿到目标类
        Class<?> targetClass = AopUtils.getTargetClass(bean);
        return bean instanceof CanalMessageListener && targetClass.isAnnotationPresent(CanalTable.class);
    }


    @Override
    public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
        conversionService = (GenericConversionService) ((DefaultListableBeanFactory) beanFactory).getConversionService();
        conversionService.addConverter(new CustomLocalDateTimeConvert());
    }

    @Data
    @Builder
    public static class CanalEventHolder {
        private CanalEntry.EventType eventType;
        private List<CanalEntry.Column> beforeColumns;
        private List<CanalEntry.Column> afterColumns;
    }

    public void setUnderlineToCamel(Boolean underlineToCamel) {
        this.underlineToCamel = underlineToCamel;
    }
}
